package com.jasonchoi.redisqueue.queue;

import lombok.extern.log4j.Log4j2;
import org.redisson.RedissonShutdownException;

import java.util.Objects;

/**
 * Redis 队列消费控制器
 * @Author: JasonChoi
 * @Date: 2020/3/16 10:25
 */
@Log4j2
public class RedisQueueHandler implements Runnable{

    private String topicKey;

    private IRedisQueueConsumer redisQueueConsumer;

    private RedisQueueUtil redisQueueUtil = RedisQueueStarter.getBean(RedisQueueUtil.class);
    private RedisQueueRetryHandler retryHandle = RedisQueueStarter.getBean(RedisQueueRetryHandler.class);

    public RedisQueueHandler(IRedisQueueConsumer redisQueueConsumer) {
        this.redisQueueConsumer = redisQueueConsumer;
        this.topicKey = redisQueueConsumer.getTopic();
    }

    @Override
    public void run() {
        while (true) {
            String tip = "RedisQueueHandler 消费【{}】：TOPIC_KEY:{} ,MESSAGE:{}";
            String errorTip = tip + ", EXCEPTION：{} ";
            Object message = null;
            try {
                message = redisQueueUtil.consume(topicKey);
                if (Objects.isNull(message)) {
                    continue;
                }
                log.debug(tip, "开始", topicKey, message);
                redisQueueConsumer.handle(message);
                log.debug(tip, "成功", topicKey, message);
            } catch (RedissonShutdownException e) {
                log.warn(e.getMessage());
            } catch (Exception e) {
                e.printStackTrace();
                log.error(errorTip, "异常", topicKey, message, e.getMessage());
                retryHandle.retryHandle(topicKey, message);
            }
        }
    }


}
